#lang racket/base
(require racket/contract
		 racket/set
		 "definition.rkt"
		 "disposable.rkt"
		 "utility.rkt")

(provide (all-defined-out))

(define/contract (subscribe obs
							[on-next void] [on-error void] [on-completed void]
							[parent-subscriber #f])
  (->* (observable?) (next/c error/c completed/c (or/c #f subscriber?)) disposable?)
  (define cus (make-custodian))
  (parameterize ([current-custodian cus])
	(let* ([dsp (disposable #f (mutable-seteq)
							(lambda (_) (custodian-shutdown-all cus)))]
		   [sub (subscriber dsp on-next on-error on-completed)]
		   [rst ((observable-on-subscribe obs) sub)])
	  (when (and (disposable? rst) (not (disposable-disposed rst)))
		(disposable-add dsp rst))
	  (when parent-subscriber
		(subscriber-add-disposable parent-subscriber dsp))
	  dsp)))

(define/contract (on-next sub x)
  (-> subscriber? any/c any)
  (unless (disposable-disposed (subscriber-disposable sub))
	((subscriber-next sub) x)))

(define/contract (on-error sub e)
  (-> subscriber? error? any)
  (unless (disposable-disposed (subscriber-disposable sub))
	((subscriber-error sub) e)
	(dispose (subscriber-disposable sub))))

(define/contract (on-completed sub)
  (-> subscriber? any)
  (unless (disposable-disposed (subscriber-disposable sub))
	((subscriber-completed sub))
	(dispose (subscriber-disposable sub))))

(define/contract (subscriber-add-disposable sub dsp)
  (-> subscriber? disposable? any)
  (unless (disposable-disposed dsp)
	(disposable-add (subscriber-disposable sub) dsp)))
